package worker

import (
	"context"
	"fmt"
	"go.etcd.io/etcd/clientv3"
	"go.etcd.io/etcd/mvcc/mvccpb"
	"go_job/common"
	"time"
)

type JobMgr struct {
	client  *clientv3.Client
	kv      clientv3.KV
	lease   clientv3.Lease
	watcher clientv3.Watcher
}

var (
	G_jobMgr *JobMgr
)

func InitJobMgr() (err error) {
	fmt.Println("etcd正在连接")
	var (
		config  clientv3.Config
		client  *clientv3.Client
		kv      clientv3.KV
		lease   clientv3.Lease
		watcher clientv3.Watcher
	)

	config = clientv3.Config{
		Endpoints:   G_config.EtcdEndPoints,
		DialTimeout: time.Duration(G_config.EtcdDialTimeOut) * time.Millisecond,
	}

	if client, err = clientv3.New(config); err != nil {
		return
	}

	kv = clientv3.NewKV(client)
	lease = clientv3.NewLease(client)
	watcher = clientv3.NewWatcher(client)

	G_jobMgr = &JobMgr{
		client:  client,
		kv:      kv,
		lease:   lease,
		watcher: watcher,
	}
	if err = G_jobMgr.WatchJobChg(); err != nil {
		return
	}

	fmt.Println("etcd链接初始化完成")

	return
}

func (jobMgr *JobMgr) WatchJobKill() (err error) {
	var (
		getResp       *clientv3.GetResponse
		job           *common.Job
		watchRevision int64
		watchChan     clientv3.WatchChan
		watchResp     clientv3.WatchResponse
		watchEvent    *clientv3.Event
		jobName       string
		jobEvent      *common.JobEvent
	)
	go func() {
		//监听下一个版本
		watchRevision = getResp.Header.Revision + 1
		watchChan = jobMgr.watcher.Watch(context.TODO(), common.JOB_KILL_DIR, clientv3.WithRev(watchRevision), clientv3.WithPrefix())

		for watchResp = range watchChan {
			for _, watchEvent = range watchResp.Events {
				switch watchEvent.Type {
				//杀死任务
				case mvccpb.PUT:
					jobName = common.ExtractJobName(string(watchEvent.Kv.Key), common.JOB_SAVE_DIR)

					job = &common.Job{
						Name: jobName,
					}
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, job)
					//推送给调度模块
					G_Scheduler.PushJobEvent(jobEvent)
					break
				//kill 过期,自动删除
				case mvccpb.DELETE:
				}
			}
		}
	}()
	return

}

func (jobMgr *JobMgr) WatchJobChg() (err error) {
	var (
		getResp       *clientv3.GetResponse
		kvPair        *mvccpb.KeyValue
		job           *common.Job
		watchRevision int64
		watchChan     clientv3.WatchChan
		watchResp     clientv3.WatchResponse
		watchEvent    *clientv3.Event
		jobName       string
		jobEvent      *common.JobEvent
	)
	//1. 获取/cron/jobs下的任务, revision
	if getResp, err = jobMgr.kv.Get(context.TODO(), common.JOB_SAVE_DIR, clientv3.WithPrefix()); err != nil {
		return
	}

	for _, kvPair = range getResp.Kvs {
		if job, err = common.Unpack(kvPair.Value); err == nil {
			jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, job)
			G_Scheduler.PushJobEvent(jobEvent)
		}
	}

	//2. 从该revision向后监听变化事件
	go func() {
		//监听下一个版本
		watchRevision = getResp.Header.Revision + 1
		watchChan = jobMgr.watcher.Watch(context.TODO(), common.JOB_SAVE_DIR, clientv3.WithRev(watchRevision), clientv3.WithPrefix())

		for watchResp = range watchChan {
			for _, watchEvent = range watchResp.Events {
				switch watchEvent.Type {
				case mvccpb.PUT:
					if job, err = common.Unpack(watchEvent.Kv.Value); err != nil {
						continue
					}
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, job)
					break
				case mvccpb.DELETE:
					//提取任务名称
					jobName = common.ExtractJobName(string(watchEvent.Kv.Key), common.JOB_SAVE_DIR)
					job = &common.Job{
						Name: jobName,
					}
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_DELETE, job)
					break
				}
				//推送给调度模块
				G_Scheduler.PushJobEvent(jobEvent)
			}
		}
	}()
	return
}

func (jobMgr *JobMgr) CreateJobLock(jobName string) (jobLock *JobLock) {
	return InitJobLock(jobMgr.kv, jobMgr.lease, jobName)
}
